package com.we.flink.mobilelabelplatform.tag;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.we.flink.mobilelabelplatform.table.TableSpecialContact;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;

public class TagSpecialContact {
    public String mobile;
    public String uid_list;

    public static Logger LOG = LoggerFactory.getLogger(TagSpecialContact.class);

    public TagSpecialContact(String mobile, String uid_list) {
        this.mobile = mobile;
        this.uid_list = uid_list;
    }

    public static class UserMobile {
        public long uid;
        public String mobile;
        public int updated_at;

        public UserMobile(long uid, String mobile, int updated_at) {
            this.uid = uid;
            this.mobile = mobile;
            this.updated_at = updated_at;
        }

        public static UserMobile init(long uid) {
            return new UserMobile(uid, "", 0);
        }

        public int getUpdated_at() {
            return updated_at;
        }
    }

    public static TagSpecialContact init(String mobile) {
        return new TagSpecialContact(mobile, "");
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableSpecialContact.QUERYSQL);
        DataStream<Tuple2<Boolean, TableSpecialContact>> inputStream =
                tableEnv.toRetractStream(tableRes, TableSpecialContact.class);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .keyBy(
                                new KeySelector<Tuple2<Boolean, TableSpecialContact>, Long>() {
                                    @Override
                                    public Long getKey(Tuple2<Boolean, TableSpecialContact> value)
                                            throws Exception {
                                        return value.f1.getUid();
                                    }
                                })
                        .flatMap(
                                new RichFlatMapFunction<
                                        Tuple2<Boolean, TableSpecialContact>,
                                        Tuple2<Boolean, TableSpecialContact>>() {
                                    ValueState<UserMobile> userMobileVState;

                                    @Override
                                    public void open(Configuration parameters) throws Exception {
                                        userMobileVState =
                                                getRuntimeContext()
                                                        .getState(
                                                                new ValueStateDescriptor<
                                                                        UserMobile>(
                                                                        "UserMobileVState",
                                                                        UserMobile.class));
                                    }

                                    @Override
                                    public void flatMap(
                                            Tuple2<Boolean, TableSpecialContact> value,
                                            Collector<Tuple2<Boolean, TableSpecialContact>> out)
                                            throws Exception {

                                        long uid = value.f1.getUid();
                                        int updated_at = value.f1.getUpdated_at();
                                        Boolean flag = value.f0;
                                        String mobile = value.f1.getMobile();

                                        UserMobile tmpState = userMobileVState.value();
                                        UserMobile curState =
                                                tmpState == null ? UserMobile.init(uid) : tmpState;

                                        System.out.println(
                                                "##flag: "
                                                        + flag
                                                        + " uid: "
                                                        + uid
                                                        + " mobile: "
                                                        + mobile
                                                        + " updated_at: "
                                                        + updated_at
                                                        + " state_update: "
                                                        + curState.getUpdated_at());

                                        /** 下推条件 */
                                        if (!flag || (updated_at > curState.getUpdated_at())) {
                                            curState.updated_at = updated_at;
                                            out.collect(value);
                                        }
                                    }
                                })
                        .keyBy(
                                new KeySelector<Tuple2<Boolean, TableSpecialContact>, String>() {
                                    @Override
                                    public String getKey(Tuple2<Boolean, TableSpecialContact> value)
                                            throws Exception {
                                        return value.f1.getMobile();
                                    }
                                })
                        .flatMap(
                                new RichFlatMapFunction<
                                        Tuple2<Boolean, TableSpecialContact>, String>() {
                                    MapState<String, Integer> mobileUidListMapState;
                                    ValueState<TagSpecialContact> mobileUidListVState;

                                    @Override
                                    public void open(Configuration parameters) throws Exception {
                                        mobileUidListMapState =
                                                getRuntimeContext()
                                                        .getMapState(
                                                                new MapStateDescriptor<
                                                                        String, Integer>(
                                                                        "mobileUidListMapState",
                                                                        String.class,
                                                                        Integer.class));
                                        mobileUidListVState =
                                                getRuntimeContext()
                                                        .getState(
                                                                new ValueStateDescriptor<
                                                                        TagSpecialContact>(
                                                                        "mobileUidListVState",
                                                                        TagSpecialContact.class));
                                    }

                                    @Override
                                    public void flatMap(
                                            Tuple2<Boolean, TableSpecialContact> value,
                                            Collector<String> out)
                                            throws Exception {
                                        Boolean flag = value.f0;
                                        String uid = String.valueOf(value.f1.getUid());
                                        String mobile = value.f1.getMobile();
                                        StringBuffer str_buf = new StringBuffer();
                                        String tmp_uid_list = "";
                                        int uids_num = 0;

                                        System.out.println(
                                                "##flatMap2## flag: "
                                                        + flag
                                                        + " uid: "
                                                        + uid
                                                        + " mobile: "
                                                        + mobile
                                                        + " update: "
                                                        + value.f1.getUpdated_at());

                                        TagSpecialContact tmpVState = mobileUidListVState.value();
                                        TagSpecialContact curVSate =
                                                (tmpVState == null
                                                        ? TagSpecialContact.init(mobile)
                                                        : tmpVState);

                                        try {
                                            /**
                                             * 针对单个uid的做计数 防止联系人来回更改
                                             * 或填写两个相同的联系人后更改其中一个导致两个相同的联系人都被清除
                                             */
                                            if (mobileUidListMapState.contains(uid)) {
                                                Integer uid_num = mobileUidListMapState.get(uid);
                                                LOG.debug(
                                                        "##DebugUse##  mobile:"
                                                                + mobile
                                                                + " uid: "
                                                                + uid
                                                                + " uid_num:"
                                                                + uid_num);
                                                if (flag) {
                                                    /** uid+1 */
                                                    mobileUidListMapState.put(uid, uid_num + 1);
                                                } else {
                                                    /** uid-1 */
                                                    if (uid_num > 1) {
                                                        mobileUidListMapState.put(uid, uid_num - 1);
                                                    } else {
                                                        mobileUidListMapState.remove(uid);
                                                    }
                                                }
                                            } else {
                                                if (flag) {
                                                    mobileUidListMapState.put(uid, 1);
                                                } else {
                                                    LOG.error(
                                                            "##Error##: mobile: "
                                                                    + mobile
                                                                    + " does not contains id: "
                                                                    + uid);
                                                }
                                            }

                                            /** in case of Map state clear the uid */
                                            if (!mobileUidListMapState.isEmpty()) {
                                                Iterator<String> keys_iterator =
                                                        mobileUidListMapState.keys().iterator();
                                                while (keys_iterator.hasNext()) {
                                                    str_buf.append(keys_iterator.next() + ",");
                                                    uids_num++;
                                                }
                                                /** trim the tail of ',' */
                                                tmp_uid_list =
                                                        str_buf.substring(0, str_buf.length() - 1);
                                            } else {
                                                uids_num = 0;
                                                tmp_uid_list = "";
                                            }
                                            LOG.debug(
                                                    "##DebugUse## mobile: "
                                                            + mobile
                                                            + " uids_num: "
                                                            + uids_num
                                                            + " uid_list: "
                                                            + tmp_uid_list);

                                            if (!curVSate.getUid_list().equals(tmp_uid_list)) {
                                                JSONObject jsonObject = new JSONObject();
                                                curVSate.uid_list = tmp_uid_list;

                                                jsonObject.put("mobile", mobile);
                                                jsonObject.put("uid_list", tmp_uid_list);
                                                out.collect(jsonObject.toString());

                                                mobileUidListVState.update(curVSate);
                                            }
                                        } catch (Exception E) {
                                            System.out.println("Exception: " + E.getMessage());
                                            LOG.error("Exception: " + E.getMessage());
                                        }
                                    }
                                });

        return out;
    }

    public String getMobile() {
        return mobile;
    }

    public String getUid_list() {
        return uid_list;
    }
}
